package com.shujia.sql

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo1WordCount {
  def main(args: Array[String]): Unit = {
    /**
      * 创建SParkSession环境
      *
      */

    val spark: SparkSession = SparkSession
      .builder()
      .appName("sql")
      .master("local")
      .config("spark.sql.shuffle.partitions", 1)//设置spark sql shuffle之后的分区数
      .getOrCreate()


    /**
      * 1、读取文件，构建DF
      *
      */

    val linesDF: DataFrame = spark
      .read
      .format("csv") //指定读取文件的格式
      .option("sep", "*") //指定分割方式
      .schema("line STRING") //指定字段名和字段类型
      .load("data/words.txt") //指定读取的路径


    /**
      * 如果要写sql 需要先注册一张表
      *
      */

    //将DF 注册成一张临时视图
    linesDF.createOrReplaceTempView("lines")

    /**
      * 写sql, 统计单词额数量
      *
      */

    val countDF: DataFrame = spark.sql(
      """
        |select
        |word,count(1) as c
        |from (
        |select
        |explode(split(line,',')) as word
        |from
        |lines) as a
        |group by word
        |
        |
      """.stripMargin)

    //show 相当于一个action算子
    //countDF.show()


    /**
      * 使用DSl 统计单词的数量
      * DSl 是一种类sql的代码
      *
      * $ 获取列的对象
      */

    //导入spark sql 所有的函数
    import org.apache.spark.sql.functions._
    //导入隐式转换
    import spark.implicits._

    val dslCountDF: DataFrame = linesDF
      .select(explode(split($"line", ",")) as "word") //将一行数据转换成多行
      .groupBy($"word") //按照单词分组, 会产生shuffle
      .count() //统计单词的数量


    dslCountDF.show()

    /**
      * 保存数据,相当于action算子
      *
      */

    dslCountDF
      .write //保存数据
      .format("csv") //指定保存数据的格式
      .mode(SaveMode.Overwrite) //指定保存模式
      .save("data/sql_count")

  }

}
